package com.atguigu.mr.table;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * @description: xxx
 * @time: 20200102 11:16
 * @author: baojinlong
 **/
public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {
    private String fileName;
    private Text nowKey = new Text();
    private TableBean nowValue = new TableBean();

    /**
     * 提前获取文件名称
     *
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        fileName = inputSplit.getPath().getName();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 获取一行
        String line = value.toString();
        // 根据表名处理数据
        String[] split = line.split("\t");
        if (fileName.startsWith("order")) {
            // 订单表处理 id pid amount 1001 01 1
            // 封装bean对象
            nowValue.setId(split[0]);
            nowValue.setPid(split[1]);
            nowValue.setAmount(Integer.parseInt(split[2]));
            nowValue.setPName("defaultValue");
            nowValue.setFlag("order");
            nowKey.set(split[1]);
        } else {
            // 产品表处理逻辑 pid pname 01 小米
            // 封装bean对象
            nowValue.setId("");
            nowValue.setPid(split[0]);
            nowValue.setAmount(0);
            nowValue.setPName(split[1]);
            nowValue.setFlag("pd");
            nowKey.set(split[0]);
        }
        // 写出
        context.write(nowKey, nowValue);
    }
}
